#pragma once

#include "Buffer.hpp"
#include "Socket.hpp"
#include "Any.hpp"

#include <vector>
#include <unordered_map>
#include <functional>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <cstring>
#include <cassert>
#include <cstdint>
#include <signal.h>
#include <unistd.h>
#include <sys/eventfd.h>
#include <sys/epoll.h>
#include <sys/timerfd.h>

#include "Log.hpp"

class EventLoop;
using EventCallBack = std::function<void()>;
class Channel
{
private:
    void Updata();
    void _SetSockfd(int sockfd)
    {
        _sockfd = sockfd;
    }
public:
    Channel(int sockfd, EventLoop *eventloop)
        : _sockfd(sockfd)
        , _eventloop(eventloop)
        , _events(0)
        , _current_events(0)
        , _read_callback(nullptr)
        , _write_callback(nullptr)
        , _error_callback(nullptr)
        , _close_callback(nullptr)
        , _event_callback(nullptr)
    {
    }
    void SetSockfd(int sockfd)
    {
        if (sockfd != -1) _SetSockfd(sockfd);
        else
            DEBUG_LOG("SetSockfd error, sockfd: %d", sockfd);
        return;
    }
    int GetSockfd()
    {
        return _sockfd;
    }
    uint32_t GetEvent() // 获取想要监控的事件
    {
        return _events;
    }

    void SetCurrentEvent(uint32_t event) // 设置实际就绪的事件
    {
        _current_events = event;
    }
    void SetReadCallback(const EventCallBack &cb)
    {
        _read_callback = cb;
    }
    void SetWriteCallback(const EventCallBack &cb)
    {
        _write_callback = cb;
    }
    void SetErrorCallback(const EventCallBack &cb)
    {
        _error_callback = cb;
    }
    void SetCloseCallback(const EventCallBack &cb)
    {
        _close_callback = cb;
    }
    void SetEventCallback(const EventCallBack &cb)
    {
        _event_callback = cb;
    }

    bool Readable() // 当前是否监控了可读
    {
        return _events & EPOLLIN;
    }
    bool Writable() // 当前是否监控了可写
    {
        return _events & EPOLLOUT;
    }
    void EnableRead() // 启动读事件监控
    {
        _events |= EPOLLIN;
        // 后边会添加到EventLoop的事件监控中
        Updata();
    }
    void EnableWrite() // 启动写事件监控
    {
        _events |= EPOLLOUT;
        // 后边会添加到EventLoop的事件监控中
        Updata();
    }
    void DisableRead() // 关闭读事件监控
    {
        _events &= ~EPOLLIN;
        // 后边会修改到EventLoop的事件监控中
        Updata();
    }
    void DisableWrite() // 关闭写事件监控
    {
        _events &= ~EPOLLOUT;
        // 后边会修改到EventLoop的事件监控中
        Updata();
    }
    void DisableAll() // 关闭所有事件监控
    {
        _events = 0;
        // 后边会修改到EventLoop的事件监控中
        Updata();
    }

    void Remove(); // 移除监控，后边会调用EventLoop接口来移除监控

    void HandleEvent() // 事件处理，一旦连接触发了事件，就调用该函数
    {
        if ((_current_events & EPOLLIN) ||
            (_current_events & EPOLLRDHUP) ||
            (_current_events & EPOLLPRI))
        {
            // 不管任何事件，都调用的回调
            if (_read_callback)
                _read_callback();
        }
        // 有可能会释放连接的操作事件，一次只处理一个
        if (_current_events & EPOLLOUT)
        {
            // 不管任何事件，都调用的回调
            if (_write_callback)
                _write_callback();
        }
        else if (_current_events & EPOLLERR)
        {
            if (_error_callback)
                _error_callback(); // 一旦出错，就会释放连接，没必要调用任意事件回调了
        }
        else if (_current_events & EPOLLHUP)
        {
            if (_close_callback)
                _close_callback();
        }
        if (_event_callback)
            _event_callback();
    }

    ~Channel()
    {
    }

private:
    int _sockfd;
    EventLoop *_eventloop;
    uint32_t _events;              // 当前需要监控的事件
    uint32_t _current_events;      // 当前连接触发的事件
    EventCallBack _read_callback;  // 可读事件被触发的回调
    EventCallBack _write_callback; // 可写事件被触发的回调
    EventCallBack _error_callback; // 错误事件被触发的回调
    EventCallBack _close_callback; // 连接断开事件被触发的回调
    EventCallBack _event_callback; // 任意事件被触发的回调
};

#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
    // 对epoll的直接操作
    void Updata(Channel *channel, int op)
    {
        int fd = channel->GetSockfd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->GetEvent();
        int n = epoll_ctl(_epfd, op, fd, &ev);
        if (n < 0)
        {
            FATAL_LOG("epoll_ctl falied!");
            abort(); // 发送信号结束程序
        }
    }
    // 判断监控事件是否已存在
    bool HasChannel(Channel *channel)
    {
        auto it = _channels.find(channel->GetSockfd());
        if (it == _channels.end())
            return false;
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(1);
        if (_epfd < 0)
        {
            FATAL_LOG("epoll_ctl falied!");
            abort(); // 发送信号结束程序
        }
    }
    // 添加/修改监控事件
    void UpdataEvent(Channel *channel)
    {
        bool flag = HasChannel(channel);
        if (!flag)
        {
            Updata(channel, EPOLL_CTL_ADD);
            _channels.insert(std::make_pair(channel->GetSockfd(), channel));
            return;
        }
        Updata(channel, EPOLL_CTL_MOD);
    }

    // 移除监控事件
    void RemoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->GetSockfd());
        if (it != _channels.end())
        {
            Updata(channel, EPOLL_CTL_DEL);
            _channels.erase(channel->GetSockfd());
        }
    }

    // 开始监控，返回活跃连接
    void Poll(std::vector<Channel*> *const active)
    {
        int n = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1); // 阻塞等待
        if (n < 0)
        {
            if (errno == EINTR)
                return;
            FATAL_LOG("epoll_wait failed!errno description: %s", strerror(errno));
            abort(); // 发送信号结束程序
        }
        for (int i = 0; i < n; i++)
        {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetCurrentEvent(_evs[i].events);
            active->emplace_back(it->second);
        }
    }
    ~Poller() {}

private:
    int _epfd;                                    // epoll的文件描述符
    struct epoll_event _evs[MAX_EPOLLEVENTS];     // 一次最大的事件数
    std::unordered_map<int, Channel*> _channels; // fd->Channel对象
};

using timer_task_t = std::function<void()>;
using release_info_t = std::function<void()>;
class TimerTask
{
public:
    TimerTask(const uint64_t id, const uint32_t delay, const timer_task_t cb)
        : _id(id)
        , _timeout(delay)
        , _task_callback(cb)
        , _canceled(false)
    {
    }

    uint32_t GetTimeout()
    {
        return _timeout;
    }

    void Cancel()
    {
        _canceled = true;
    }

    void SetRelease(const release_info_t cb)
    {
        _release_info = cb;
    }

    ~TimerTask()
    {
        if (!_canceled)
            _task_callback();
        _release_info();
    }

private:
    uint64_t _id;                 // 定时器任务对象id，方便在时间轮中查找
    uint32_t _timeout;            // 定时任务的超时时间
    bool _canceled;               // 是否被取消
    timer_task_t _task_callback;  // 定时器对象执行的回调任务
    release_info_t _release_info; // 删除timewheel中保存的定时器对象信息
};

class TimeWheel
{
private:
    using task_ptr = std::shared_ptr<TimerTask>;
    using task_weakptr = std::weak_ptr<TimerTask>;
    void RemoveTimerTask(uint64_t id)
    {
        auto pos = _timers.find(id);
        if (pos != _timers.end())
        {
            _timers.erase(pos);
        }
    }
    
    static int CreatTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            FATAL_LOG("timerfd_create failed!");
            abort();
        }
        // int timerfd_settime(int fd, int flags, const struct itimerspec *new_value, struct itimerspec *old_value);
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0; // 第一次超时后，每次超时的时间间隔
        timerfd_settime(timerfd, 0, &itime, nullptr);
        return timerfd;
    }

    uint64_t ReadTimerfd()
    {
        uint64_t times;
        // 有可能因为其他描述符的事件处理花费时间比较长，然后在处理定时器描述符事件的时候，有可能就已经超时了很多次
        ssize_t n = ::read(_timerfd, &times, sizeof times);
        if (n < 0)
        {
            FATAL_LOG("timerfd read failed!");
            abort();
        }
        return times;
    }
    // 这个函数应该被每秒钟执行一次，相当于秒钟向后走了一步
    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear(); // 清空指定位置的数组，就会把数组中保存的所有管理定时器对象的shared_ptr释放掉
    }
    void OnTime()
    {
        // 根据实际的超时次数，执行实际的定时任务次数
        uint64_t times = ReadTimerfd();
        for (uint64_t i = 0; i < times; i++)
            RunTimerTask();
    }

    void AddTimerTaskInLoop(const uint64_t id, const timer_task_t cb, const uint32_t delay)
    {
        // if (delay > _capacity)
        //     return false; // 超出最大延迟时间
        task_ptr tp = std::make_shared<TimerTask>(id, delay, cb); // 提高内存分配效率，同时也是异常安全的
        tp->SetRelease(std::bind(&TimeWheel::RemoveTimerTask, this, id));
        uint32_t pos = (_tick + delay) % _capacity;
        _wheel[pos].emplace_back(tp);
        _timers.insert(std::make_pair(id, task_weakptr(tp)));
        // return true;
    }
    void RefreshTimerTaskInLoop(const uint64_t id)
    {
        // 通过保存的定时器对象的task_weakptr来构造一个task_ptr出来，添加到时间轮中
        auto it = _timers.find(id);
        if (it == _timers.end())
            return;                       // 没找到定时任务
        task_ptr tmp = it->second.lock(); // lock获取weak_ptr管理的对象对应的shared_ptr
        uint32_t delay = tmp->GetTimeout();
        uint32_t pos = (_tick + delay) % _capacity;
        _wheel[pos].emplace_back(tmp);
    }
    void CancelTimerTaskInLoop(const uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
            return;                 // 找不到这个id（错误id）
        task_ptr tmp = it->second.lock(); // lock获取weak_ptr管理的对象对应的shared_ptr
        if (tmp) tmp->Cancel();
    }

public:
    TimeWheel(EventLoop *loop)
        : _tick(0), _capacity(60), _wheel(_capacity), _timerfd(CreatTimerfd()), _eventloop(loop), _timerfd_channel(std::make_unique<Channel>(_timerfd, _eventloop)) // C++14
    {
        _timerfd_channel->SetReadCallback(std::bind(&TimeWheel::OnTime, this));
        _timerfd_channel->EnableRead(); // 启动读事件监控
    }

    // 有线程安全问题--这个接口不能被外界使用者调用，只能在模块内的EventLoop线程内执行
    bool HasTimerTask(const uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
            return false; // 找不到这个定时任务
        return true;
    }
    // 定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题
    // 如果不想加锁,那就把对定时器的所有操作,都放到一个线程中进行
    void AddTimerTask(const uint64_t id, const timer_task_t cb, const uint32_t delay);
    void RefreshTimerTask(const uint64_t id);
    void CancelTimerTask(const uint64_t id);

    ~TimeWheel() {}

private:
    uint32_t _tick;                                     // 当前秒针，走到哪执行哪
    uint32_t _capacity;                                 // 最大延迟时间
    std::vector<std::vector<task_ptr>> _wheel;          // 用于管理定时器任务的数据结构
    std::unordered_map<uint64_t, task_weakptr> _timers; // 用id来映射weakptr指向的对象，这里很重要！
    int _timerfd;                                       // 定时器描述符--可读事件回调就是读取计数器，执行定时任务
    EventLoop *_eventloop;
    std::unique_ptr<Channel> _timerfd_channel;
};

using Functor = std::function<void()>;
class EventLoop
{
private:
    static int CreatEventfd()
    {
        int efd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            FATAL_LOG("eventfd failed!");
            abort(); // 程序异常退出
        }
        return efd;
    }

    // 用于判断当前线程是否是EventLoop对应的线程
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    void WakeUpEventfd()
    {
        uint64_t val = 1;
        int ret = ::write(_event_fd, &val, sizeof val);
        if (ret < 0)
        {
            if (errno == EINTR)
                return;
            ERROR_LOG("write eventfd failed!");
            abort();
        }
    }

    void ReadEventfd()
    {
        uint64_t res = 0;
        int ret = ::read(_event_fd, &res, sizeof res);
        if (ret < 0)
        {
            // EINTR被信号打断，EAGAIN无数据可读
            if (errno == EINTR || errno == EAGAIN)
                return;
            ERROR_LOG("read eventfd failed!");
            abort();
        }
    }
public:
    EventLoop()
        : _thread_id(std::this_thread::get_id())
        , _event_fd(CreatEventfd())
        , _event_channel(std::make_unique<Channel>(_event_fd, this)) // C++14
        , _timewheel(this)
    {
        // 给eventfd添加可读事件监控，读取eventfd事件通知次数
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        // 启动eventfd的读事件监控
        _event_channel->EnableRead();
    }
    
    void QueueLoop(const Functor &cb) // 将操作压入任务池
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(cb);
        }
        // 唤醒有可能因为没有事件就绪，而导致的epoll阻塞；
        // 其实就是给eventfd写入一个数据，eventfd就会触发可读事件
        WakeUpEventfd();
    }
    
    void AssertInLoop()
    {
        assert(IsInLoop());
    }

    // 判断将要执行的任务是否处于当前线程中，如果在则执行，否则压入队列
    void RunInLoop(const Functor &cb)
    {
        if (IsInLoop())
        {
            cb();
            return;
        }
        QueueLoop(cb);
    }

    void UpdataEvent(Channel *channel) // 用于添加/修改描述符的事件监控
    {
        _poll.UpdataEvent(channel);
    }
    void RemoveEvent(Channel *channel) // 用于移除描述符的事件监控
    {
        _poll.RemoveEvent(channel);
    }

    // 执行任务池中的所有任务
    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            functor.swap(_tasks);
        }
        for (const auto &f : functor)
            f();
    }
    void Start() // 三步走：事件监控->就绪事件处理->执行任务
    {
        while (true)
        {
            // 事件监控
            std::vector<Channel*> actives;
            _poll.Poll(&actives);
            // 就绪事件处理
            for (const auto &channel : actives)
                channel->HandleEvent();
            // 执行任务
            RunAllTask();
        }
    }

    bool HasTimerTask(const uint64_t id)
    {
        return _timewheel.HasTimerTask(id);
    }
    void AddTimerTask(const uint64_t id, const timer_task_t cb, const uint32_t delay)
    {
        _timewheel.AddTimerTask(id, cb, delay);
    }
    void RefreshTimerTask(const uint64_t id)
    {
        _timewheel.RefreshTimerTask(id);
    }
    void CancelTimerTask(const uint64_t id)
    {
        _timewheel.CancelTimerTask(id);
    }

    ~EventLoop() {}

private:
    std::thread::id _thread_id; // 线程id
    int _event_fd;              // eventfd唤醒IO事件监控有可能导致阻塞
    std::unique_ptr<Channel> _event_channel;
    Poller _poll;                // 进行所有描述符监控
    TimeWheel _timewheel;  //定时器模块
    std::vector<Functor> _tasks; // 任务池
    std::mutex _mutex;           // 实现任务池操作的线程安全
};

void Channel::Updata()
{
    _eventloop->UpdataEvent(this);
}
void Channel::Remove()
{
    _eventloop->RemoveEvent(this);
}

void TimeWheel::AddTimerTask(const uint64_t id, const timer_task_t cb, const uint32_t delay)
{
    _eventloop->RunInLoop(std::bind(&TimeWheel::AddTimerTaskInLoop, this, id, cb, delay));
}
void TimeWheel::RefreshTimerTask(const uint64_t id)
{
    _eventloop->RunInLoop(std::bind(&TimeWheel::RefreshTimerTaskInLoop, this, id));
}
void TimeWheel::CancelTimerTask(const uint64_t id)
{
    _eventloop->RunInLoop(std::bind(&TimeWheel::CancelTimerTaskInLoop, this, id));
}

typedef enum
{
    DISCONNECTED, // 连接关闭状态
    CONNECTING, // 连接建立成功，待处理状态
    CONNECTED, // 连接建立完成，各种设置已完成，可通信状态
    DISCONNECTING // 连接待关闭状态
}ConnStatu;

class Connection : public std::enable_shared_from_this<Connection>
{
public:
    // 这四个回调函数，是让服务器模块来设置的
    using connection_ptr = std::shared_ptr<Connection>;
    using ConnectedCallback = std::function<void(const connection_ptr&)>;
    using MessageCallback = std::function<void(const connection_ptr&, Buffer*)>;
    using ClosedCallback = std::function<void(const connection_ptr&)>;
    using AnyEventCallback = std::function<void(const connection_ptr&)>;
private:
    // 五个channel的事件回调函数
    // 描述符可读事件触发后调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callback
    void HandleRead()
    {
        // 1.接收socket数据，放到缓冲区
        std::string buf;
        buf.resize(65536); // 64K
        ssize_t n = _socket.NonBlockRecv(&buf);
        if (n < 0) // 出错了
        {
            ShutDownInLoop(); // 不能直接关闭连接
            return;
        }
        buf.resize(n); // 这里一定要，因为加入收了125字节，但是write buf的时候大小是整个buf大小64k
        // 将数据放入输入缓冲区
        _in_buffer.Write(buf);
        // 2.调用_message_callback，进行业务处理
        if (_in_buffer.GetReadableSize() > 0)
        {
            // shared_from_this从当前对象自身获取自身的shared_ptr管理对象
            _message_callback(shared_from_this(), &_in_buffer);
            return;
        }
    }
    // 描述符可写事件触发后调用的函数，将发送缓冲区中的数据进行发送
    void HandleWrite()
    {
        ssize_t n = _socket.NonBlockSend(std::string(_out_buffer.GetReadAddr(), _out_buffer.GetReadableSize()));
        if (n < 0)
        {
            // 发送错误，就该关闭连接，但要先判断输入缓冲区是否还有数据
            if (_in_buffer.GetReadableSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);  
            }
            Release(); // 实际的关闭释放操作
            return;
        }
        _out_buffer.MoveBackReadOffset(n); // 千万要记得，将读偏移向后移动，因为NonBlockSend的是一个string临时对象
        if (_out_buffer.GetReadableSize() == 0)
        {
            _channel.DisableWrite(); // 没有数据待发送了，关闭写事件监控
            // 如果当前是连接待关闭状态，则有数据，发送玩数据释放连接，没有数据则直接释放
            if (_status == DISCONNECTING)
            {
                Release();
            }
        }
    }
    // 描述符触发挂断事件后调用的函数
    void HandleClose()
    {
        // 一旦连接挂断，套接字就什么都干不了了，因此有数据待处理就处理，完毕就关闭连接
        if (_in_buffer.GetReadableSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);  
        }
        Release(); // 实际的关闭释放操作
        return;
    }
    // 描述符触发任意事件后调用的函数:1.刷新连接的活跃度--延迟定时销毁任务；2.调用组件使用者的任意事件回调
    void HandleEvent()
    {
        if (!_enable_inactive_release)  return;
        _eloop->RefreshTimerTask(_connection_id);
        if (_event_callback) _event_callback(shared_from_this());
    }
    // 描述符触发错误事件调用的函数
    void HandleError()
    {
        HandleClose();
    }
    // 连接获取之后，所处的状态下要进行各种设置（给channel设置事件回调，启动读监控，调用回调）
    void EstablishedInLoop()
    {
        // 1.修改连接状态；2.启动读事件监控；3.调用回调函数
        assert(_status == CONNECTING); // 当前的状态必须是上层的半连接状态
        _status = CONNECTED; // 当前函数执行完毕，则连接进入已完成连接状态
        // 一旦启动读事件监控，就有可能立即触发读事件，如果这时候启动了非活跃连接销毁
        _channel.EnableRead();
        if (_connected_callback) _connected_callback(shared_from_this());
    }
    // 这个接口并不是实际的发送数据，而是把数据放到了发送缓冲区，启动了可读事件监控
    void SendInLoop(const Buffer& buf)
    {
        if (_status == DISCONNECTED) return;
        _out_buffer.Write(buf);
        if (_channel.Writable() == false)
        {
            _channel.EnableWrite();
        }
    }
    // 这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
    void ShutDownInLoop()
    {
        _status = DISCONNECTING; // 设置连接为半关闭状态
        if (_in_buffer.GetReadableSize() > 0)
        {
            if (_message_callback) _message_callback(shared_from_this(), &_in_buffer);
        }
        // 要么就是写入数据的时候出错关闭，要么就是没有待发送数据，直接关闭
        if (_out_buffer.GetReadableSize() > 0)
        {
            if (_channel.Writable() == false)
            {
                _channel.EnableWrite();
            }
        }
        else if (_out_buffer.GetReadableSize() == 0) Release();
    }
    // 实际的释放接口
    void ReleaseInLoop()
    {
<<<<<<< HEAD
        // 0. 检查对象是否存活
=======
	        // 0. 检查对象是否存活
>>>>>>> 5aed3a43e71bf34703df67cd31f7c7a7d6427a65
        auto wp = weak_from_this();
        if (wp.expired()) {
            DEBUG_LOG("Connection %lu already destroyed, skip ReleaseInLoop", _connection_id);
            return;
        }
        // 1.修改连接状态；2.移除连接的事件监控；3.关闭描述符；4.如果当前定时器队列中还有定时销毁任务，则取消任务
        _status = DISCONNECTED;
        _channel.Remove();
        _socket.Close();
        if (_eloop->HasTimerTask(_connection_id)) CancelInactiveReleaseInLoop();
        // 5.调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection被释放，再去处理会出错
        if (_closed_callback) _closed_callback(shared_from_this());
        // 移除服务器内部管理的连接信息
        if (_server_closed_callback) _server_closed_callback(shared_from_this());
    }
    // 启动非活跃连接超时释放
    void EnableInactiveReleaseInLoop(uint32_t sec)
    {
        // 1.将判断标志_enable_inactive_release置true
        _enable_inactive_release = true;
        // 2.如果当前定时销毁任务已经存在，那就刷新延迟一下即可
        if (_eloop->HasTimerTask(_connection_id))
        {
            _eloop->RefreshTimerTask(_connection_id);
            return;
        }
        // 3.如果不存在定时销毁任务，则添加定时销毁任务
        _eloop->AddTimerTask(_connection_id, std::bind(&Connection::Release, this), sec);
    }
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if (_eloop->HasTimerTask(_connection_id))
            _eloop->CancelTimerTask(_connection_id);
    }
    void UpgradeInLoop(const Any& context, const ConnectedCallback& conn, const MessageCallback& msg, 
        const ClosedCallback& closed, const AnyEventCallback& event)
    {
        SetContext(context);
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }
public:
    Connection(EventLoop* eloop, uint64_t connection_id, int sockfd)
     : _connection_id(connection_id)
     , _sockfd(sockfd)
     , _enable_inactive_release(false)
     , _eloop(eloop)
     , _status(CONNECTING)
     , _socket(sockfd)
     , _channel(sockfd, eloop)
     , _connected_callback(nullptr)
     , _message_callback(nullptr)
     , _closed_callback(nullptr)
     , _event_callback(nullptr)
     , _server_closed_callback(nullptr)
    {
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
    }
    int GetSockfd()
    {
        return _sockfd;
    }
    uint64_t GetConnectionid()
    {
        return _connection_id;
    }

    // 连接建立就绪后，进行channel回调设置，启动读监控，调用_connected_callback
    void Established()
    {
        _eloop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }

    bool IsConnected() // 是否处于CONNECTED状态
    {
        return _status == CONNECTED;
    }

    void SetContext(const Any& context) // 设置上下文--连接建立完成时
    {
        _context = context;
    }
    Any* GetContext() // 获取上下文
    {
        return &_context;
    }
    void SetConnectedCallback(const ConnectedCallback& cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback& cb)
    {
        _closed_callback = cb;
    }
    void SetServerClosedCallback(const ClosedCallback& cb)
    {
        _server_closed_callback = cb;
    }
    void SetAnyEventedCallback(const AnyEventCallback& cb)
    {
        _event_callback = cb;
    }

    // 将数据发送到发送缓冲区，启动写事件监控
    void Send(const char* const data, size_t len)
    {
        // 外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入任务池，有可能并没有被执行
        // 因此有可能执行的时候，data指向的空间有可能已经被释放了
        Buffer buf;
        buf.Write(data);
        _eloop->RunInLoop(std::bind(&Connection::SendInLoop, this, buf));
    }
    void ShutDown() // 提供给组件使用者的关闭接口，并不实际关闭，需要判断有没有数据待处理
    {
        _eloop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    void Release()
    {
        _eloop->QueueLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
    void EnableInactiveRelease(uint32_t sec) // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    {
        _eloop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    void CancelInactiveRelease()
    {
        _eloop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 协议切换，重置上下文，以及阶段性处理回调--非线程安全(这个接口必须在EventLoop线程中立即执行)
    // 防备新的事件触发后，处理的时候，切换任务还没有被执行，会导致数据使用原协议处理
    void Upgrade(const Any& context, const ConnectedCallback& conn, const MessageCallback& msg, 
                 const ClosedCallback& closed, const AnyEventCallback& event)
    {
        _eloop->AssertInLoop();
        _eloop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
    ~Connection()
    {
        DEBUG_LOG("Release Connection:%p", this);
    }
private:
    uint64_t _connection_id; // 连接的唯一id，便于查找和管理
    // uint64_t _timetask_id; // 定时器id，必须是唯一的，这块为了简化操作，使用_conn_id作为定时器id
    int _sockfd; // 连接关联的文件描述符
    bool _enable_inactive_release; // 判断连接是否启动非活跃连接销毁的标志，默认为false
    EventLoop* _eloop; // 连接所关联的一个EventLoop
    ConnStatu _status; // 连接状态
    TcpSocket _socket; // 套接字的管理
    Channel _channel; // 连接的事件管理
    Buffer _in_buffer; // 输入缓冲区--存放从socket中读取到的数据
    Buffer _out_buffer; // 输出缓冲区--存放要发送给对端的数据
    Any _context; // 请求的接受处理上下文
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    ClosedCallback _server_closed_callback; // 组件内的连接关闭回调--组件内设置的
};

class Acceptor
{
using AcceptCallback = std::function<void(int)>;
private:
    // 监听套接字的读事件回调处理函数 -- 获取新连接，调用回调
    void HandleRead()
    {
        InetAddr clientaddr;
        int newfd = _socket.Accept(&clientaddr);
        if (newfd < 0) return;
        if (_accept_callback) _accept_callback(newfd);
    }
public:
    // 不能将启动读事件监控添加到构造函数中，必须在设置回调函数后，再去启动，否则
    // 有可能造成启动监控后，立即有事件，处理的时候，回调函数还没设置：新连接得不到处理，且资源泄漏
    Acceptor(EventLoop* eloop, uint16_t port)
    : _socket(-1)
    , _eloop(eloop)
    , _channel(-1, eloop)
    , _accept_callback(nullptr)
    {
        _socket.BulidListenSocket(port); // 必须让_sockfd构造完毕，虚函数填入虚函数表才可以进行下面的操作
        _channel.SetSockfd(_socket.GetSockfd());
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }
    void SetAcceptCallback(const AcceptCallback& cb)
    {
        _accept_callback = cb;
    }
    void Listen()
    {
        _channel.EnableRead();
    }
    ~Acceptor() {}
private:
    TcpSocket _socket; // 用于创建监听套接字
    EventLoop* _eloop; // 用于对监听套接字进行事件监控
    Channel _channel; // 用于对监听套接字进行事件管理
    AcceptCallback _accept_callback;
};

class ThreadLoop
{
private:
    // 实例化EventLoop对象，唤醒_cond上有可能阻塞的线程，并且运行EventLoop模块的工功能
    void ThreadEntry()
    {
        EventLoop eloop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _eloop = &eloop;
            _cond.notify_all();
        }
        eloop.Start();
    }
public:
    // 创建线程，设定线程入口函数
    ThreadLoop()
     : _eloop(nullptr)
     , _thread(std::thread(&ThreadLoop::ThreadEntry, this))
    {}
    EventLoop* GetEventLoop() // 返回当前线程关联的EventLoop对象指针
    {
        EventLoop* eloop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&] () {
                return _eloop != nullptr;
            });
            eloop = _eloop;
        }
        return eloop;
    }
    ~ThreadLoop() {}
private:
    EventLoop* _eloop;
    // 用于实现_eloop获取的同步关系，避免线程创建了，但是_eloop还没有实例化之前去获取_eloop
    std::mutex _mutex;
    std::condition_variable _cond;
    std::thread _thread;
};

class ThreadLoopPool
{
public:
    ThreadLoopPool(EventLoop* eloop)
    : _thread_count(0)
    , _next_idx(0)
    , _base_eloop(eloop)
    {}
    EventLoop* GetNextEventLoop()
    {
        if (_thread_count == 0)
            return _base_eloop;
        _next_idx = (_next_idx + 1) % _thread_count;
        return _eloops[_next_idx];
    }
    void SetThreadCount(uint32_t count)
    {
        _thread_count = count;
    }
    void Run()
    {
        if (_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _eloops.resize(_thread_count);
            for (int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new ThreadLoop();
                _eloops[i] = _threads[i]->GetEventLoop(); // 主线程将其二者绑定，会出现线程安全，因为要先new出来才能再绑定，有同步关系
            }
        }
    }
    ~ThreadLoopPool() {}
private:
    uint32_t _thread_count; // 可设置的线程数
    uint32_t _next_idx;     // 下一次的_threads下标，用于实现时间片轮转算法（Round-Robin）的rr轮转
    EventLoop* _base_eloop; // 主从Reactor线程中的主线程，当_thread_count为0时，它一个人处理所有任务
    std::vector<ThreadLoop*> _threads;
    std::vector<EventLoop*> _eloops;
};

class TcpServer
{
private:
    void NewConnection(int newfd) // 为新连接构造一个Connection进行管理
    {
        _next_id++;
        Connection::connection_ptr connection = std::make_shared<Connection>(_thread_loop_pool.GetNextEventLoop()
                                                , _next_id, newfd);
        connection->SetConnectedCallback(_connected_callback);
        connection->SetMessageCallback(_message_callback);
        connection->SetClosedCallback(_closed_callback);
        connection->SetAnyEventedCallback(_event_callback);
        connection->SetServerClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if (_enable_inactive_release)
            connection->EnableInactiveRelease(_timeout); // 启动非活跃超时销毁
        connection->Established(); // 就绪初始化
        _connections.insert(std::make_pair(_next_id, connection));
    }
    void RemoveConnectionInLoop(const Connection::connection_ptr& conn)
    {
        int id = conn->GetConnectionid();
        auto it = _connections.find(id);
        if (it == _connections.end()) return;
        _connections.erase(it);
    }
    void RemoveConnection(const Connection::connection_ptr& conn) // 从管理Connection的_connections中移除连接信息
    {
        _base_eloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }
    void ScheduledTimerTaskInLoop(const Functor& task, uint32_t sec)
    {
        _next_id++;
        _base_eloop.AddTimerTask(_next_id, task, sec);
    }
public:
    using ConnectedCallback = std::function<void(const Connection::connection_ptr&)>;
    using MessageCallback = std::function<void(const Connection::connection_ptr&, Buffer*)>;
    using ClosedCallback = std::function<void(const Connection::connection_ptr&)>;
    using AnyEventCallback = std::function<void(const Connection::connection_ptr&)>;
    TcpServer(uint16_t port)
    : _next_id(0)
    , _port(port)
    , _enable_inactive_release(false)
    , _acceptor(&_base_eloop, _port)
    , _thread_loop_pool(&_base_eloop)
    , _connected_callback(nullptr)
    , _message_callback(nullptr)
    , _closed_callback(nullptr)
    , _event_callback(nullptr)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen(); // 将监听套接字挂到base_eloop上开始监听事件
    }
    void SetThreadCount(uint32_t count)
    {
        _thread_loop_pool.SetThreadCount(count);
    }
    void SetConnectedCallback(const ConnectedCallback& cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback& cb)
    {
        _closed_callback = cb;
    }
    void SetAnyEventedCallback(const AnyEventCallback& cb)
    {
        _event_callback = cb;
    }
    void EnableInactiveRelease(uint32_t sec)
    {
        _timeout = sec;
        _enable_inactive_release = true;
    }
    using Functor = std::function<void()>;
    void ScheduledTimerTask(const Functor& task, uint32_t sec) // 用于添加一个定时任务
    {
        _base_eloop.RunInLoop(std::bind(&TcpServer::ScheduledTimerTaskInLoop, this, task, sec));
    }
    void Start()
    {
        _thread_loop_pool.Run(); // 启动线程池中的从属线程
        _base_eloop.Start();
    }
    ~TcpServer() {}
private:
    uint64_t _next_id; // 自动连续增长的连接ID，定时任务和连接id都是他
    uint16_t _port;
    uint32_t _timeout; // 非活跃连接的统计时间，多长时间无通信就是非活跃连接
    bool _enable_inactive_release;//是否启动了非活跃连接超时销毁的判断标志
    EventLoop _base_eloop;  //这是主线程的EventLoop对象，负责监听事件的处理
    Acceptor _acceptor;    //这是监听套接字的管理对象
    ThreadLoopPool _thread_loop_pool; // 从属EventLoop线程池
    std::unordered_map<uint64_t, Connection::connection_ptr> _connections;//保存管理所有连接对应的shared_ptr对象
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
};

class NetWork
{
public:
    NetWork()
    {
        DEBUG_LOG("NetWork Init success!");
        signal(SIGPIPE, SIG_IGN);
    }
};

static NetWork network;
